package com.atguigu.day11;

import com.atguigu.bean.Top2;
import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableAggregateFunction;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

public class FlinkSQL19_Function_UDTAF {

    public static void main(String[] args) throws Exception {

        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //2.读取端口数据并转换为JavaBean
        SingleOutputStreamOperator<WaterSensor> waterSensorDS = env.socketTextStream("hadoop102", 9999)
                .map(line -> {
                    String[] fields = line.split(",");
                    return new WaterSensor(fields[0],
                            Long.parseLong(fields[1]),
                            Double.parseDouble(fields[2]));
                });

        //3.将流转换为动态表
        Table table = tableEnv.fromDataStream(waterSensorDS);

        //4.使用UDTF
        //4.1 TableAPI：不注册使用
        Table select1 = table.groupBy($("id"))
                .flatAggregate(call(Top2Vc.class, $("vc")).as("rk", "vc"))
                .select($("id"), $("rk"), $("vc"));

        //4.2 TableAPI：注册使用
//        tableEnv.createTemporarySystemFunction("top2", Top2Vc.class);
//        Table select2 = table.groupBy($("id"))
//                .flatAggregate(call("top2", $("vc")).as("rk", "vc"))
//                .select($("id"), $("rk"), $("vc"));

        //5.转换为流进行打印
        tableEnv.toRetractStream(select1, Row.class).print("TableAPI:不注册使用>>>>>>>>>");
//        tableEnv.toRetractStream(select2, Row.class).print("TableAPI:注册使用>>>>>>>>>>>");

        //6.启动
        env.execute();
    }

    public static class Top2Vc extends TableAggregateFunction<Tuple2<String, Double>, Top2> {

        @Override
        public Top2 createAccumulator() {
            return new Top2(Double.MIN_VALUE, Double.MIN_VALUE);
        }

        public void accumulate(Top2 acc, Double value) {
            if (value > acc.getTop2()) {//大于第二
                if (value > acc.getTop1()) {//大于第一
                    acc.setTop2(acc.getTop1());
                    acc.setTop1(value);
                } else {
                    acc.setTop2(value);
                }
            }

            System.out.println(acc);
        }

        public void emitValue(Top2 acc, Collector<Tuple2<String, Double>> out) {
            System.out.println("emitValue:" + acc);
            out.collect(new Tuple2<>("Top1", acc.getTop1()));
            if (acc.getTop2() > Double.MIN_VALUE) {
                out.collect(new Tuple2<>("Top2", acc.getTop2()));
            }
        }
    }
}
